package net.xo68.destiny.starter.worker.handler.impl;

import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.contract.domain.StringRespondResult;
import net.xo68.destiny.contract.domain.report.JobHandleReplyDatagram;
import net.xo68.destiny.core.discovery.NodeDiscoveryClient;
import net.xo68.destiny.core.rpc.RpcClient;
import net.xo68.destiny.core.util.ThreadUtils;
import net.xo68.destiny.starter.worker.SchedulerWorkerProperties;
import net.xo68.destiny.starter.worker.handler.JobHandleReplyFailStore;
import net.xo68.destiny.starter.worker.handler.JobHandleReplyProvider;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 任务处理过程回复实现
 * @author wuxie
 * @version 2020-11-4
 */
@Slf4j
@Component
public class JobHandleReplyProviderImpl implements JobHandleReplyProvider, ApplicationContextAware, SmartInitializingSingleton, DisposableBean {

    private static ApplicationContext applicationContext=null;

    /**
     * 初始化
     */
    public static final int REPLY_STATE_INIT=0;
    /**
     * 已经启动
     */
    public static final int REPLY_STATE_STARTED=1;
    /**
     * 关闭的
     */
    public static final int REPLY_STATE_SHUTDOWN=2;

    /**
     * 回复状态
     */
    private final AtomicInteger replyState=new AtomicInteger(REPLY_STATE_INIT);
    /**
     * 重试延时记数，为0是重试，失败时增加，成功是置0
     */
    private final AtomicInteger retryDelayCount=new AtomicInteger(0);

    private final SchedulerWorkerProperties schedulerWorkerProperties;
    private final NodeDiscoveryClient nodeDiscoveryClient;
    private final RpcClient rpcClient;
    private final ScheduledExecutorService replyExecutorService;
    private final LinkedBlockingQueue<JobHandleReplyDatagram> eventQueue;
    private final JobHandleReplyFailStore replyFailStore;

    public JobHandleReplyProviderImpl(SchedulerWorkerProperties schedulerWorkerProperties,NodeDiscoveryClient nodeDiscoveryClient,RpcClient rpcClient,JobHandleReplyFailStore replyFailStore) {
        this.schedulerWorkerProperties=schedulerWorkerProperties;
        this.nodeDiscoveryClient=nodeDiscoveryClient;
        this.rpcClient=rpcClient;
        this.replyFailStore=replyFailStore;

        this.replyExecutorService = ThreadUtils.newThreadScheduled("JobHandleReply",this.schedulerWorkerProperties.getHandleReplyThreadCount());
        this.eventQueue = new LinkedBlockingQueue<>();

    }

    @Override
    public boolean push(JobHandleReplyDatagram replyDatagram) {
        return this.eventQueue.offer(replyDatagram);
    }

    @Override
    public void run() {
        final int maxDrain=schedulerWorkerProperties.getHandleReplyBatchMax();
        while (replyState.get()==REPLY_STATE_STARTED){
            final List<JobHandleReplyDatagram> eventList = new ArrayList<>(maxDrain);
            int numDrained = this.eventQueue.drainTo(eventList, maxDrain);
            if (numDrained == 0)
                break;
            //通过httpclient上传处理报文 this.replyExecutorService.submit();
            //log.info("eventList: {}",eventList);
            StringRespondResult respondResult=rpcClient.tryPostJson(nodeDiscoveryClient.getManageHosts(),null,"/api/jobState", StringRespondResult.class,eventList);
            if(respondResult==null || respondResult.isFail()){
                log.error("上报数据失败，{}",eventList);
                //将失败的持久存储，以便重试上报
                replyFailStore.batchAdd(eventList);
                incrementRetryDelayCount();
            }
        }
        if(replyState.get()==REPLY_STATE_STARTED){
            //todo 对上传失败的报文进行重试处理
            if(retryDelayCount.get() < 1){
                final List<JobHandleReplyDatagram> failList =replyFailStore.limit(maxDrain);
                //log.debug("回复重试，待重试的记录数量: {}", failList.size());
                if(failList!=null && failList.size()>0){
                    StringRespondResult respondResult=rpcClient.tryPostJson(nodeDiscoveryClient.getManageHosts(),null,"/api/jobState", StringRespondResult.class,failList);
                    if(respondResult!=null && respondResult.isSuccess()){
                        log.debug("重试成功，failList: {}", failList);
                        //将重试成功的移除
                        replyFailStore.remove(failList);
                        //有回复失败记录并重试成功时，下个周期继续重试
                        retryDelayCount.set(0);
                    }else {
                        incrementRetryDelayCount();
                    }
                }else {
                    //没有回复失败记录，延时最大时间后，再重试
                    retryDelayCount.set(schedulerWorkerProperties.getHandleReplyMaxDelay());
                }
            }else {
                retryDelayCount.decrementAndGet();
            }

        }
    }

    /**
     * 回复失败时增加 重试延时记数
     */
    private void incrementRetryDelayCount(){
        //失败后增加指定步长的重试延时记录数
        retryDelayCount.addAndGet(15);
        if(retryDelayCount.get()>schedulerWorkerProperties.getHandleReplyMaxDelay()){
            retryDelayCount.set(schedulerWorkerProperties.getHandleReplyMaxDelay());
        }
    }

    @Override
    public void destroy() throws Exception {
        if(replyState.compareAndSet(REPLY_STATE_STARTED,REPLY_STATE_SHUTDOWN)){
            this.replyExecutorService.shutdown();

            //将未上传的报文 持久化保存
            final List<JobHandleReplyDatagram> eventList = new ArrayList<>();
            int numDrained = this.eventQueue.drainTo(eventList);
            if(numDrained>0){
                //将未上传的报文 持久化保存
                replyFailStore.batchAdd(eventList);
            }
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if(replyState.compareAndSet(REPLY_STATE_INIT,REPLY_STATE_STARTED)){
            this.replyExecutorService.scheduleWithFixedDelay(this,0,this.schedulerWorkerProperties.getHandleReplyDelayMilli(), TimeUnit.MILLISECONDS);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        JobHandleReplyProviderImpl.applicationContext=applicationContext;
    }
}
